package com.gin.monitor.agg.warn;

import com.gin.monitor.agg.connector.kafka.FlinkKafkaUtils;
import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.DangerousDrivingWarning;
import com.gin.monitor.agg.vo.MonitorInfo;
import com.gin.monitor.agg.vo.OutOfLimitSpeedInfo;
import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;


/**
 * cep 危险驾驶
 * <p>
 * cd /opt/software/kafka_2.11-2.0.0/bin
 * bin/kafka-server-start.sh -daemon config/server.properties
 * ./kafka-console-consumer.sh --bootstrap-server node01:9092 --topic t_traffic_monitor --group group01 --property print.key=true --property print.value=true --property key.separator=,
 * 测试生产
 * ./kafka-console-producer.sh --broker-list node01:9092 --topic t_traffic_monitor
 * <p>
 * 测试时注意: 使用当前时间戳
 * 1620637883000,0002,75312,京L22188,60.1,27,31
 * 1620637884000,0002,75312,京L22188,70.1,27,31
 * 1620637885000,0002,75312,京L22188,80.1,27,31
 * 1620637885000,0002,75312,京L22188,80.1,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class DangerousDriverWarningAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //默认最近开始读取
        //DataStreamSource<TrafficInfo> streamKafka = env.addSource(FlinkKafkaUtils.getKafkaConsumer());
        //从第一行开始读取. 测试用!
        SingleOutputStreamOperator<OutOfLimitSpeedInfo> streamKafka = env
                .addSource(FlinkKafkaUtils.getKafkaConsumer().setStartFromEarliest())
                .map(new RichMapFunction<TrafficInfo, OutOfLimitSpeedInfo>() {

                    Connection conn = null;
                    PreparedStatement pst = null;
                    HashMap<String, MonitorInfo> monitorMap = new HashMap<>();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = DriverManager.getConnection("jdbc:mysql://node01/test", "root", "123456");
                        pst = conn.prepareStatement("select monitor_id,road_id,speed_limit,area_id from t_monitor_info where speed_limit > 0");
                        ResultSet set = pst.executeQuery();
                        while (set.next()) {
                            //缓存卡口信息
                            MonitorInfo info = new MonitorInfo(set.getString(1), set.getString(2), set.getInt(3), set.getString(4));
                            monitorMap.put(info.monitorId, info);
                        }
                        set.close();
                        pst.close();
                        conn.close();
                    }

                    @Override
                    public OutOfLimitSpeedInfo map(TrafficInfo in) throws Exception {
                        //首先从Map集合中判断是否存在卡口的限速，如果不存在，默认限速为60
                        MonitorInfo info = monitorMap.get(in.monitorId);
                        info = info == null ? new MonitorInfo(in.monitorId, in.roadId, 60, in.areaId) : info;
                        return new OutOfLimitSpeedInfo(in.car, in.monitorId, in.roadId, in.speed, info.limitSpeed, in.actionTime);
                    }
                })
                //指定时间戳字段, 及允许延时时间
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OutOfLimitSpeedInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));

        //streamKafka.print();

        //匹配规则
        Pattern<OutOfLimitSpeedInfo, OutOfLimitSpeedInfo> pattern = Pattern.<OutOfLimitSpeedInfo>begin("begin")
                .where(new IterativeCondition<OutOfLimitSpeedInfo>() {
                    @Override
                    public boolean filter(OutOfLimitSpeedInfo info, Context<OutOfLimitSpeedInfo> context) {
                        //超速20%
                        return info.limitSpeed * 1.2 < info.realSpeed;
                    }
                })
                //超速3次或者3次以上
                .timesOrMore(3)
                .greedy()
                //定义时间范围是2分钟内
                .within(Time.minutes(2));

        PatternStream<OutOfLimitSpeedInfo> ps = CEP.pattern(streamKafka
                        .keyBy((KeySelector<OutOfLimitSpeedInfo, String>) outOfLimitSpeedInfo -> outOfLimitSpeedInfo.car),
                pattern);

        ps.select((PatternSelectFunction<OutOfLimitSpeedInfo, DangerousDrivingWarning>) map -> {
            List<OutOfLimitSpeedInfo> outOfLimitSpeedList = map.get("begin");
            StringBuilder sb = new StringBuilder();
            sb.append("该车辆涉嫌危险驾驶,");
            int i = 1;

            for (OutOfLimitSpeedInfo info : outOfLimitSpeedList) {
                sb.append("第").append(i).append("个经过卡口是:").append(info.monitorId).append(" -->");
                i += 1;
            }
            return new DangerousDrivingWarning(outOfLimitSpeedList.get(0).car, sb.toString(), System.currentTimeMillis(), 0.0D);
        }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
